package com.ganqiang1983.dbdiff;

import java.util.List;
import java.util.concurrent.BlockingQueue;

import org.apache.log4j.Logger;

import com.ganqiang1983.dbdiff.ScheduleManager.JOB_TYPE;

/**
 * 生产者
 * 主要用于从配置文件中生产Job
 */
public class Scheduler
{
  private static Logger logger = Logger.getLogger(Scheduler.class);
  private BlockingQueue<Job> dpool;
  private BlockingQueue<Job> spool;
  private BlockingQueue<Job> cpool;
  private Integer djobsize = 0;
  private Integer sjobsize = 0;
  private Integer cjobsize = 0;

  public Scheduler(JOB_TYPE type, BlockingQueue<Job> dpool,
      BlockingQueue<Job> spool, BlockingQueue<Job> cpool)
  {
    this.dpool = dpool;
    this.spool = spool;
    this.cpool = cpool;
  }

  public void run()
  {
    logger.info("[Main-Thread] Scheduler start");
    try {
      doDump();
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      logger.error("[Main-Thread] Scheduler being interrupted : " + e.getMessage());
    }
    logger.info("[Main-Thread] Scheduler end");
  }

  /**
   * step 1 : 生成job，为后阶段准备
   * @throws InterruptedException
   */
  private void doDump() throws InterruptedException
  {
    // 1.将内存中的配置分解成多对pair，即加载多个job到流水线（queue）中
    JobCollector jc = JobCollector.newInstance();
    List<Job> jobList = null;
    try {
      logger.info("[Main-Thread] Scheduler begin to create job.");
      jobList = jc.getJobList();
      logger.info("[Main-Thread] Scheduler create job completed.");
    } catch (Exception e) {
      logger.error("[Main-Thread] Scheduler get jobs failure : " + e.getMessage(), e);
      throw new InterruptedException("[Dump-Main-Thread] Scheduler get jobs occur error : " + e.getMessage());
    }

    // 2.如果需要执行undo job，那么可以让程序接着做上次未完成的job
    // 因为从jobfile中获取出来的job肯定是最新状态的job，所以不用删除jobfile文件.
    int newsize = jobList.size();
    if (ScheduleManager.newInstance().IS_PROCEED_MODE) {
      jobList = jc.getUndoJob(jobList);
      int size = jobList.size();
      logger.info("[Main-Thread] Scheduler initialize the [proceed mode] and ready to execute [" + size + "] unfinished job");
      // 判断未完成的job列表是否为0,是的话说明之前的job已全部完成，则需抛异常
      if (size == 0) {
        InterruptedException e = new InterruptedException("[Dump-Main-Thread] Scheduler"
            + " found all the job are already done,please switch [start mode] in order to recompare");
        logger.error("[Main-Thread] Scheduler found all the job are already done,please switch [start mode] in order to recompare", e);
        throw e;
      }
    } else {
      logger.info("[Main-Thread] Scheduler initialize the [start mode] and ready to execute [" + newsize + "] new job");
      // 判断job列表
      if (newsize == 0) {
        InterruptedException e = new InterruptedException("[Dump-Main-Thread] Scheduler did not find any new job,please check the configuration file");
        logger.error("[Main-Thread] Scheduler did not find any new job,please check the configuration file", e);
        throw e;
      }
    }

    // 3.查看pair目录是否已经存在不然就新生成一个目录
    boolean flag = FileMaker.mkDiffDir();
    if (flag) {
      logger.info("[Main-Thread] Scheduler create a new dbdiff directory : [" + DBDiffConfUtil.DIFF_PATH + "]");
    } else {
      logger.info("[Main-Thread] Scheduler override the old dbdiff directory : [" + DBDiffConfUtil.DIFF_PATH + "]");
    }

    // 4.依次将job加入到D_QUEUE队列中
    try {
      for (Job job : jobList) {
        if (job.getType().equals(JOB_TYPE.DUMP)) {
          djobsize = djobsize + 1;
          dpool.put(job);
          logger.info("[Main-Thread] Scheduler add a pair of [" + job.getPairId()
              + "] to the dump queue");
        } else if (job.getType().equals(JOB_TYPE.SORT)) {
          sjobsize = sjobsize + 1;
          spool.put(job);
          logger.info("[Main-Thread] Scheduler add a pair of [" + job.getPairId()
              + "] to the sort queue");
        } else {
          cjobsize = cjobsize + 1;
          cpool.put(job);
          logger.info("[Main-Thread] Scheduler add a pair of [" + job.getPairId()
              + "] to the diff queue");
        }
      }
      sjobsize = sjobsize + djobsize;
      cjobsize = cjobsize + sjobsize;
      logger.info("[Main-Thread] Scheduler assign job to dump is [" + djobsize + "] and sort is [" + sjobsize + "] and compare is [" + cjobsize + "]");
    } catch (InterruptedException e) {
      logger.error("[Main-Thread] Scheduler assign jobs to queue failure : " + e.getMessage(), e);
      throw new InterruptedException("[Main-Thread] Scheduler assign jobs to queue failure : " + e.getMessage());
    }
  }

  public int getDjobsize()
  {
    return djobsize;
  }

  public int getSjobsize()
  {
    return sjobsize;
  }

  public int getCjobsize()
  {
    return cjobsize;
  }

}
